package com.shujia.onhbase

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.{HConnection, HConnectionManager, HTableInterface, Put}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2WiterPartition {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local").setAppName("onhabse")

    val sc: SparkContext = new SparkContext(conf)


    val linesRDD: RDD[String] = sc.textFile("spark/data/words.txt")

    val wordsRDD: RDD[String] = linesRDD.flatMap(_.split(","))

    val countRDD: RDD[(String, Int)] = wordsRDD.map((_, 1)).reduceByKey(_ + _)


    /**
      * 将数据保存到hbase中  以单词作为rowkey
      *
      * create 'wc','info'
      *
      */


    /**
      * foreach 每一条数据都会创建hbase的连接,会导致产生大量的连接对象,  会出现OOM
      *
      */

    /*countRDD.foreach(kv => {
      val word: String = kv._1
      val count: Int = kv._2


      //1  创建hbase连接
      val configuration: Configuration = new Configuration
      configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181")
      val connection: HConnection = HConnectionManager.createConnection(configuration)

      //2  获取表
      val table: HTableInterface = connection.getTable(TableName.valueOf("wc"))


      //3 插入数据

      val put: Put = new Put(word.getBytes())
      put.add("info".getBytes(), "num".getBytes(), count.toString.getBytes())

      table.put(put)

      //4  关闭连接
      connection.close()

    })*/

    /**
      * foreachPartition  每一个分区创建一个连接
      *
      */


    countRDD.foreachPartition(iter => {

      //1  创建hbase连接
      val configuration: Configuration = new Configuration
      configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181")
      val connection: HConnection = HConnectionManager.createConnection(configuration)

      //2  获取表
      val table: HTableInterface = connection.getTable(TableName.valueOf("wc"))


      //遍历一个分区中的所有数据
      iter.foreach(kv => {
        val word: String = kv._1
        val count: Int = kv._2
        //3 插入数据

        val put: Put = new Put(word.getBytes())
        put.add("info".getBytes(), "num".getBytes(), count.toString.getBytes())

        table.put(put)
      })


      // 4  关闭连接
      connection.close()

    })

  }
}
